package cn.xxgc.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
object WordCount01 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new
        SparkConf().setAppName("spark").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))//每隔5秒划分一
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop101",9999)
    val resultDS: DStream[(String,Int)] = lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _)
    resultDS.print()
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(stopSparkContext = true,stopGracefully = true) //优雅关闭
  }
}
